#include "join_hashtable.hpp"
#include "arrow_help.hpp"
#include "common_macro.hpp"
#include "spdlog/spdlog.h"
#include <cmath>

namespace {

double RoundToDecimalPlaces(double value, int precision) {
    double factor = std::pow(10.0, precision);
    return std::round(value * factor) / factor;
}

float RoundToDecimalPlaces(float value, int precision) {
    float factor = std::pow(10.0f, precision);
    return std::round(value * factor) / factor;
}

}

namespace DaseX {

JoinHashTable::JoinHashTable(std::vector<int> build_ids,
                             std::vector<int> probe_ids,
                             std::vector<ExpressionTypes> comparison_types,
                             JoinTypes join_type)
                             : build_ids(std::move(build_ids)),
                             probe_ids(std::move(probe_ids)),
                             comparison_types(comparison_types),
                             join_type(join_type) {
    NUM_PARTITIONS = NumberOfPartitions(radix_bits);
    SHIFT = Shift(radix_bits);
    MASK = Mask(radix_bits);
    buckets.resize(NUM_PARTITIONS);
    // 初始化 buckets，包含NUM_PARTITIONS个TupleBucket对象，后续优化，可以使用惰性初始化
    for (auto &bucket : buckets) {
        bucket = std::make_shared<TupleBucket>();
    }
}

void JoinHashTable::Build(std::shared_ptr<arrow::RecordBatch> &chunk) {
    if(unlikely(schema == nullptr)) {
        schema = chunk->schema();
        switch (join_type) {
            case JoinTypes::LEFT:
            case JoinTypes::RIGHT:
            {
                GetSingleNullRow();
                break;
            }
            default:
                break;
        } // switch
    }
    int num_rows = chunk->num_rows();
    if(num_rows == 0) {
        return;
    }
    data_chunks.push_back(chunk);
    total_count += num_rows;
    std::vector<std::shared_ptr<arrow::Array>> join_keys;
    if(comparison_types.empty()) {
        for (auto &build_id : build_ids) {
            std::shared_ptr<arrow::Array> join_key_p = chunk->column(build_id);
            join_keys.emplace_back(join_key_p);
        }
    } else {
        for (int i = 0; i < comparison_types.size(); i++) {
            if(comparison_types[i] == ExpressionTypes::COMPARE_EQUAL) {
                std::shared_ptr<arrow::Array> join_key_p = chunk->column(build_ids[i]);
                join_keys.emplace_back(join_key_p);
            }
        }
    }

    // 计算Hash值
    std::vector<size_t> hashes = Hashs(join_keys, num_rows);
    // 计算每行数据应该属于那个桶
    std::vector<int> bucket_idx = ComputeBucketIndices(hashes, num_rows);
    // 将对应数据散列到各个桶
    std::vector<std::shared_ptr<arrow::Array>> join_keys_p;
    for(auto build_id : build_ids){
        std::shared_ptr<arrow::Array> join_key = chunk->column(build_id);
        join_keys_p.emplace_back(join_key);
    }
    ScatterData(join_keys_p, bucket_idx, num_rows, hashes);
}

std::vector<size_t> JoinHashTable::Hashs(std::vector<std::shared_ptr<arrow::Array>> &join_keys, int count) {
    std::vector<size_t> hashes(count);
    for(int k = 0; k < count; k++) {
        std::vector<size_t> hashs_t;
        int column_size = join_keys.size();
        for(int i = 0; i < column_size; i++) {
            auto &join_key = join_keys[i];
            switch (join_key->type_id()) {
                case arrow::Type::INT32:
                {
                    std::hash<int> intHasher;
                    auto int32array = std::static_pointer_cast<arrow::Int32Array>(join_key);
                    int val = int32array->Value(k);
                    size_t hash_p = intHasher(val);
                    hashs_t.emplace_back(hash_p);
                    break;
                }
                case arrow::Type::FLOAT:
                {
                    std::hash<float> float32Hasher;
                    auto float32array = std::static_pointer_cast<arrow::FloatArray>(join_key);
                    float val = float32array->Value(k);
                    float val_p = RoundToDecimalPlaces(val, precision);
                    size_t hash_p = float32Hasher(val_p);
                    hashs_t.emplace_back(hash_p);
                    break;
                }
                case arrow::Type::DOUBLE:
                {
                    std::hash<double> doubleHasher;
                    auto doubleArray = std::static_pointer_cast<arrow::DoubleArray>(join_key);
                    double val = doubleArray->Value(k);
                    double val_p = RoundToDecimalPlaces(val, precision);
                    size_t hash_p = doubleHasher(val_p);
                    hashs_t.emplace_back(hash_p);
                    break;
                }
                case arrow::Type::STRING:
                {
                    std::hash<std::string> stringHasher;
                    auto stringArray = std::static_pointer_cast<arrow::StringArray>(join_key);
                    std::string val = stringArray->GetString(k);
                    size_t hash_p = stringHasher(val);
                    hashs_t.emplace_back(hash_p);
                    break;
                }
                default:
                    throw std::runtime_error("Not Supported For Arrow Type!!!");
            } // switch
        } // for
        size_t hash_tmp = 0;
        for(int i = 0; i < column_size; i++) {
            hash_tmp = hash_tmp ^ (hashs_t[i] << i);
        }
        hashes[k] = hash_tmp;
    } // for
    return hashes;
} // Hashs

std::vector<int> JoinHashTable::ComputeBucketIndices(std::vector<size_t> &join_key_hash, int count) {
    std::vector<int> bucket_idx(count);
    for(int i = 0; i < count; i++) {
        size_t hash = join_key_hash[i];
        bucket_idx[i] = hash & (NUM_PARTITIONS - 1);
    }
    return bucket_idx;
}

void JoinHashTable::ScatterData(std::vector<std::shared_ptr<arrow::Array>> &join_keys, std::vector<int> &bucket_idx, int count, std::vector<size_t> &hashs) {
    std::vector<std::vector<std::shared_ptr<Value>>> keys(count);
    for(int i = 0; i < join_keys.size(); i++){
        auto &join_key = join_keys[i];
        switch (join_key->type_id()) {
            case arrow::Type::INT32:
            {
                auto int32array = std::static_pointer_cast<arrow::Int32Array>(join_key);
                for(int j = 0; j < count; j++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(int32array->Value(j));
                    keys[j].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::FLOAT:
            {
                auto floatArray = std::static_pointer_cast<arrow::FloatArray>(join_key);
                for(int j = 0; j < count; j++) {
                    float val = RoundToDecimalPlaces(floatArray->Value(j), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[j].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::DOUBLE:
            {
                auto doubleArray = std::static_pointer_cast<arrow::DoubleArray>(join_key);
                for(int j = 0; j < count; j++) {
                    double val = RoundToDecimalPlaces(doubleArray->Value(j), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[j].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::STRING:
            {
                auto stringArray = std::static_pointer_cast<arrow::StringArray>(join_key);
                for(int j = 0; j < count; j++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(stringArray->GetString(j));
                    keys[j].emplace_back(val_p);
                }
                break;
            }
            default:
                throw std::runtime_error("Invalid arrow::Type!!!");
        } // switch
    }
    for(int i = 0; i < count; i++) {
        auto &bucket = buckets[bucket_idx[i]];
        std::shared_ptr<EntrySingle> entry_p = std::make_shared<EntrySingle>(keys[i], Offset(nums_chunk, i), hashs[i]);
        bucket->InsertEntry(entry_p);
    }
    nums_chunk++;
}

std::shared_ptr<ProbeState> JoinHashTable::Probe(std::shared_ptr<arrow::RecordBatch> &chunk) {
    if(total_count == 0) {
        return std::make_shared<ProbeState>();
    }
    int num_rows = chunk->num_rows();
    std::vector<std::shared_ptr<arrow::Array>> join_keys;
    if(comparison_types.empty()) {
        for(auto probe_id : probe_ids){
            std::shared_ptr<arrow::Array> join_key = chunk->column(probe_id);
            join_keys.emplace_back(join_key);
        }
    } else {
        for (int i = 0; i < comparison_types.size(); i++) {
            if(comparison_types[i] == ExpressionTypes::COMPARE_EQUAL) {
                std::shared_ptr<arrow::Array> join_key_p = chunk->column(probe_ids[i]);
                join_keys.emplace_back(join_key_p);
            }
        }
    }
    // 计算Hash值
    std::vector<size_t> hashes = Hashs(join_keys, num_rows);
    // 计算每行数据应该属于那个桶
    std::vector<int> bucket_idx = ComputeBucketIndices(hashes, num_rows);
    // 获取探测结果
    std::shared_ptr<ProbeState> probe_state;
    switch (join_type) {
        case JoinTypes::SEMI:
        {
            std::vector<std::shared_ptr<arrow::Array>> join_keys_p;
            for(auto probe_id : probe_ids){
                std::shared_ptr<arrow::Array> join_key = chunk->column(probe_id);
                join_keys_p.emplace_back(join_key);
            }
            probe_state = GatherSemiData(join_keys_p, bucket_idx, num_rows, hashes);
            break;
        }
        case JoinTypes::ANTI:
        {
            std::vector<std::shared_ptr<arrow::Array>> join_keys_p;
            for(auto probe_id : probe_ids){
                std::shared_ptr<arrow::Array> join_key = chunk->column(probe_id);
                join_keys_p.emplace_back(join_key);
            }
            probe_state = GatherAntiData(join_keys_p, bucket_idx, num_rows, hashes);
            break;
        }
        default:
            probe_state = GatherData(join_keys, bucket_idx, num_rows, hashes);
    } // switch
    return probe_state;
}

std::shared_ptr<ProbeState> JoinHashTable::GatherData(std::vector<std::shared_ptr<arrow::Array>> &join_keys,
                                                      std::vector<int> &bucket_idx,
                                                      int count,
                                                      std::vector<size_t> &hashes) {
    std::shared_ptr<ProbeState> probe_state = std::make_shared<ProbeState>();
    std::vector<std::vector<std::shared_ptr<Value>>> keys(count);
    for(int j = 0; j < join_keys.size(); j++){
        auto &join_key = join_keys[j];
        switch (join_key->type_id()) {
            case arrow::Type::INT32:
            {
                auto int32array = std::static_pointer_cast<arrow::Int32Array>(join_key);
                for(int i = 0; i < count; i++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(int32array->Value(i));
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::FLOAT:
            {
                auto floatArray = std::static_pointer_cast<arrow::FloatArray>(join_key);
                for(int i = 0; i < count; i++) {
                    float val = RoundToDecimalPlaces(floatArray->Value(i), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::DOUBLE:
            {
                auto doubleArray = std::static_pointer_cast<arrow::DoubleArray>(join_key);
                for(int i = 0; i < count; i++) {
                    double val = RoundToDecimalPlaces(doubleArray->Value(i), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::STRING:
            {
                auto stringArray = std::static_pointer_cast<arrow::StringArray>(join_key);
                for(int i = 0; i < count; i++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(stringArray->GetString(i));
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            default:
                throw std::runtime_error("Invalid arrow::Type!!!");
        } // switch
    } // for
    for(int i = 0; i < count; i++) {
        auto &bucket = buckets[bucket_idx[i]];
        int tuple_nums = bucket->tuple_nums;
        std::vector<Offset> res;
        bool is_match = false;
        size_t hash_p = hashes[i];
        for(int j = 0; j < tuple_nums; j++) {
            auto entry_single = bucket->GetEntry(j);
            bool flag = true; //是否所有字段都满足
            if(hash_p != entry_single->hash_val) {
                flag = false;
            } else {
                for(int k = 0; k < join_keys.size(); k++) {
                    if(*(keys[i][k]) != *(entry_single->keys[k])) {
                        flag = false;
                        break;
                    }
                } // for
            }
            if(flag){
                is_match = true;
                Offset offset = entry_single->val;
                res.emplace_back(offset);
            }
        } // for
        if(is_match) {
            probe_state->AddMatch(i);
            probe_state->AddMatchResult(i, res);
            probe_state->bit_map.push_back(1);
        } else {
            probe_state->bit_map.push_back(0);
        }
    } // for
    return probe_state;
}

std::shared_ptr<ProbeState> JoinHashTable::GatherSemiData(std::vector<std::shared_ptr<arrow::Array>> &join_keys,
                                                      std::vector<int> &bucket_idx,
                                                      int count,
                                                      std::vector<size_t> &hashes) {
    std::shared_ptr<ProbeState> probe_state = std::make_shared<ProbeState>();
    std::vector<std::vector<std::shared_ptr<Value>>> keys(count);
    for(int j = 0; j < join_keys.size(); j++){
        auto &join_key = join_keys[j];
        switch (join_key->type_id()) {
            case arrow::Type::INT32:
            {
                auto int32array = std::static_pointer_cast<arrow::Int32Array>(join_key);
                for(int i = 0; i < count; i++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(int32array->Value(i));
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::FLOAT:
            {
                auto floatArray = std::static_pointer_cast<arrow::FloatArray>(join_key);
                for(int i = 0; i < count; i++) {
                    float val = RoundToDecimalPlaces(floatArray->Value(i), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::DOUBLE:
            {
                auto doubleArray = std::static_pointer_cast<arrow::DoubleArray>(join_key);
                for(int i = 0; i < count; i++) {
                    double val = RoundToDecimalPlaces(doubleArray->Value(i), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::STRING:
            {
                auto stringArray = std::static_pointer_cast<arrow::StringArray>(join_key);
                for(int i = 0; i < count; i++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(stringArray->GetString(i));
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            default:
                throw std::runtime_error("Invalid arrow::Type!!!");
        } // switch
    } // for
    for(int i = 0; i < count; i++) {
        auto &bucket = buckets[bucket_idx[i]];
        int tuple_nums = bucket->tuple_nums;
        std::vector<Offset> res;
        bool is_match = false;
        size_t hash_p = hashes[i];
        for(int j = 0; j < tuple_nums; j++) {
            auto entry_single = bucket->GetEntry(j);
            bool flag = true; //是否所有字段都满足
            if(hash_p != entry_single->hash_val) {
                flag = false;
            } else {
                for(int k = 0; k < join_keys.size(); k++) {
                    if(comparison_types.empty()) {
                        if(*(keys[i][k]) != *(entry_single->keys[k])) {
                            flag = false;
                            break;
                        }
                    } else {
                        // <, >, !=, =, >=, <=
                        switch (comparison_types[k]) {
                            case ExpressionTypes::COMPARE_LESSTHAN:
                            {
                                if(*(keys[i][k]) > *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_GREATERTHAN:
                            {
                                if(*(keys[i][k]) <= *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_NOTEQUAL:
                            {
                                if(*(keys[i][k]) == *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_EQUAL:
                            {
                                if(*(keys[i][k]) != *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_GREATERTHANOREQUALTO:
                            {
                                if(*(keys[i][k]) < *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_LESSTHANOREQUALTO:
                            {
                                if(*(keys[i][k]) > *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            default:
                                throw std::runtime_error("Invalid Compare Type!!!");
                        } // switch
                        if(flag == false)
                            break;
                    }
                } // for
            }
            if(flag){
                is_match = true;
                Offset offset = entry_single->val;
                res.emplace_back(offset);
                break;
            }
        } // for
        if(is_match) {
            probe_state->AddMatch(i);
            probe_state->AddMatchResult(i, res);
            probe_state->bit_map.push_back(1);
        } else {
            probe_state->bit_map.push_back(0);
        }
    } // for
    return probe_state;
}

std::shared_ptr<ProbeState> JoinHashTable::GatherAntiData(std::vector<std::shared_ptr<arrow::Array>> &join_keys,
                                                      std::vector<int> &bucket_idx,
                                                      int count,
                                                      std::vector<size_t> &hashes) {
    std::shared_ptr<ProbeState> probe_state = std::make_shared<ProbeState>();
    std::vector<std::vector<std::shared_ptr<Value>>> keys(count);
    for(int j = 0; j < join_keys.size(); j++){
        auto &join_key = join_keys[j];
        switch (join_key->type_id()) {
            case arrow::Type::INT32:
            {
                auto int32array = std::static_pointer_cast<arrow::Int32Array>(join_key);
                for(int i = 0; i < count; i++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(int32array->Value(i));
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::FLOAT:
            {
                auto floatArray = std::static_pointer_cast<arrow::FloatArray>(join_key);
                for(int i = 0; i < count; i++) {
                    float val = RoundToDecimalPlaces(floatArray->Value(i), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::DOUBLE:
            {
                auto doubleArray = std::static_pointer_cast<arrow::DoubleArray>(join_key);
                for(int i = 0; i < count; i++) {
                    double val = RoundToDecimalPlaces(doubleArray->Value(i), precision);
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(val);
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            case arrow::Type::STRING:
            {
                auto stringArray = std::static_pointer_cast<arrow::StringArray>(join_key);
                for(int i = 0; i < count; i++) {
                    std::shared_ptr<Value> val_p = std::make_shared<Value>(stringArray->GetString(i));
                    keys[i].emplace_back(val_p);
                }
                break;
            }
            default:
                throw std::runtime_error("Invalid arrow::Type!!!");
        } // switch
    } // for
    for(int i = 0; i < count; i++) {
        auto &bucket = buckets[bucket_idx[i]];
        int tuple_nums = bucket->tuple_nums;
        std::vector<Offset> res;
        bool is_match = false;
        size_t hash_p = hashes[i];
        for(int j = 0; j < tuple_nums; j++) {
            auto entry_single = bucket->GetEntry(j);
            bool flag = true; //是否所有字段都满足
            if(hash_p != entry_single->hash_val) {
                flag = false;
            } else {
                for(int k = 0; k < join_keys.size(); k++) {
                    if(comparison_types.empty()) {
                        if(*(keys[i][k]) != *(entry_single->keys[k])) {
                            flag = false;
                            break;
                        }
                    } else {
                        // <, >, !=, =, >=, <=
                        switch (comparison_types[k]) {
                            case ExpressionTypes::COMPARE_LESSTHAN:
                            {
                                if(*(keys[i][k]) > *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_GREATERTHAN:
                            {
                                if(*(keys[i][k]) <= *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_NOTEQUAL:
                            {
                                if(*(keys[i][k]) == *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_EQUAL:
                            {
                                if(*(keys[i][k]) != *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_GREATERTHANOREQUALTO:
                            {
                                if(*(keys[i][k]) < *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            case ExpressionTypes::COMPARE_LESSTHANOREQUALTO:
                            {
                                if(*(keys[i][k]) > *(entry_single->keys[k])) {
                                    flag = false;
                                }
                                break;
                            }
                            default:
                                throw std::runtime_error("Invalid Compare Type!!!");
                        } // switch
                        if(flag == false)
                            break;
                    }
                } // for
            }
            if(flag){
                is_match = true;
                Offset offset = entry_single->val;
                res.emplace_back(offset);
            }
        } // for
        if(is_match) {
            probe_state->AddMatch(i);
            probe_state->AddMatchResult(i, res);
            probe_state->bit_map.push_back(1);
        } else {
            probe_state->bit_map.push_back(0);
        }
    } // for
    return probe_state;
}

template<JoinTypes join_type>
void JoinHashTable::GetJoinResult(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    throw std::runtime_error("Unknown JoinType");
}

template<>
void JoinHashTable::GetJoinResult<JoinTypes::INNER>(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    auto &match = probe_state->match;
    if(match.empty()) {
        return;
    }
    auto &match_result = probe_state->match_result;
    std::vector<std::shared_ptr<arrow::RecordBatch>> results_p;
    for(int i = 0; i < match.size(); i++) {
        std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(match[i], 1);
        std::vector<Offset> &match_offset = match_result[match[i]];
        int match_size = match_offset.size();
        for(int j = 0; j < match_size; j++) {
            std::shared_ptr<arrow::RecordBatch> right = GetSingleRow(match_offset[j]);
            std::shared_ptr<arrow::RecordBatch> res_p;
            // TODO: 参数可以提取到类中，后续优化
            Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
            results_p.push_back(res_p);
        }
    }
    result = arrow::ConcatenateRecordBatches(results_p).ValueOrDie();
}

template<>
void JoinHashTable::GetJoinResult<JoinTypes::LEFT>(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    auto &match = probe_state->match;
    auto &match_result = probe_state->match_result;
    std::vector<std::shared_ptr<arrow::RecordBatch>> results_p;
    for(int i = 0; i < match.size(); i++) {
        std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(match[i], 1);
        std::vector<Offset> &match_offset = match_result[match[i]];
        int match_size = match_offset.size();
        for(int j = 0; j < match_size; j++) {
            std::shared_ptr<arrow::RecordBatch> right = GetSingleRow(match_offset[j]);
            std::shared_ptr<arrow::RecordBatch> res_p;
            // TODO: 参数可以提取到类中，后续优化
            Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
            results_p.push_back(res_p);
        }
    }
    if(last_probe_state != nullptr) {
        auto &bit_map = probe_state->bit_map;
        auto &last_bit_map = last_probe_state->bit_map;
        int size_map = bit_map.size();
        if(!last_bit_map.empty()) {
            for(int i = 0; i < size_map; i++) {
                bit_map[i] |= last_bit_map[i];
            }
        }
        std::vector<int> no_match;
        if(is_final_probe) {
            for(int i = 0; i < size_map; i++) {
                if(bit_map[i] == 0) {
                    no_match.push_back(i);
                }
            }
            int no_match_size = no_match.size();
            if(no_match_size != 0) {
                for(int j = 0; j < no_match_size; j++) {
                    std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(no_match[j], 1);
                    std::shared_ptr<arrow::RecordBatch> right = null_row->Slice(0, 1);
                    std::shared_ptr<arrow::RecordBatch> res_p;
                    // TODO: 参数可以提取到类中，后续优化
                    Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
                    results_p.push_back(res_p);
                }
            } // if(no_match_size != 0)
        } // if(is_final_probe)
    } // if(last_probe_state != nullptr)
    else {
        if(is_final_probe) {
            auto &bit_map = probe_state->bit_map;
            int size_map = bit_map.size();
            std::vector<int> no_match;
            for(int i = 0; i < size_map; i++) {
                if(bit_map[i] == 0) {
                    no_match.push_back(i);
                }
            }
            int no_match_size = no_match.size();
            if(no_match_size != 0) {
                for(int j = 0; j < no_match_size; j++) {
                    std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(no_match[j], 1);
                    std::shared_ptr<arrow::RecordBatch> right = null_row->Slice(0, 1);
                    std::shared_ptr<arrow::RecordBatch> res_p;
                    // TODO: 参数可以提取到类中，后续优化
                    Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
                    results_p.push_back(res_p);
                }
            } // if(no_match_size != 0)
        }
    }
    if(results_p.empty()) {
        return;
    }
    result = arrow::ConcatenateRecordBatches(results_p).ValueOrDie();
}

// Right JOIN实现逻辑同LEFT，只有Build和Probe的表序不同，Right JOIN必须右表为Probe表，而LEFT JOIN必须左表为Probe表
template<>
void JoinHashTable::GetJoinResult<JoinTypes::RIGHT>(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    auto &match = probe_state->match;
    auto &match_result = probe_state->match_result;
    std::vector<std::shared_ptr<arrow::RecordBatch>> results_p;
    for(int i = 0; i < match.size(); i++) {
        std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(match[i], 1);
        std::vector<Offset> &match_offset = match_result[match[i]];
        int match_size = match_offset.size();
        for(int j = 0; j < match_size; j++) {
            std::shared_ptr<arrow::RecordBatch> right = GetSingleRow(match_offset[j]);
            std::shared_ptr<arrow::RecordBatch> res_p;
            // TODO: 参数可以提取到类中，后续优化
            Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
            results_p.push_back(res_p);
        }
    }
    if(last_probe_state != nullptr) {
        auto &bit_map = probe_state->bit_map;
        auto &last_bit_map = last_probe_state->bit_map;
        int size_map = bit_map.size();
        for(int i = 0; i < size_map; i++) {
            bit_map[i] |= last_bit_map[i];
        }
        std::vector<int> no_match;
        if(is_final_probe) {
            for(int i = 0; i < size_map; i++) {
                if(bit_map[i] == 0) {
                    no_match.push_back(i);
                }
            }
            int no_match_size = no_match.size();
            if(no_match_size != 0) {
                for(int j = 0; j < no_match_size; j++) {
                    std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(no_match[j], 1);
                    std::shared_ptr<arrow::RecordBatch> right = null_row->Slice(0, 1);
                    std::shared_ptr<arrow::RecordBatch> res_p;
                    // TODO: 参数可以提取到类中，后续优化
                    Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
                    results_p.push_back(res_p);
                }
            } // if(no_match_size != 0)
        } // if(is_final_probe)
    } // if(last_probe_state != nullptr)
    else {
        if(is_final_probe) {
            auto &bit_map = probe_state->bit_map;
            int size_map = bit_map.size();
            std::vector<int> no_match;
            for(int i = 0; i < size_map; i++) {
                if(bit_map[i] == 0) {
                    no_match.push_back(i);
                }
            }
            int no_match_size = no_match.size();
            if(no_match_size != 0) {
                for(int j = 0; j < no_match_size; j++) {
                    std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(no_match[j], 1);
                    std::shared_ptr<arrow::RecordBatch> right = null_row->Slice(0, 1);
                    std::shared_ptr<arrow::RecordBatch> res_p;
                    // TODO: 参数可以提取到类中，后续优化
                    Util::join_two_record_batch(row, row->num_columns(), 0,right, right->num_columns(), 0, res_p);
                    results_p.push_back(res_p);
                }
            } // if(no_match_size != 0)
        }
    }
    if(results_p.empty()) {
        return;
    }
    result = arrow::ConcatenateRecordBatches(results_p).ValueOrDie();
}

template<>
void JoinHashTable::GetJoinResult<JoinTypes::OUTER>(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    throw std::runtime_error("Unknown JoinType");
}

// 内部表为Build表，返回Probe为true的外部表字段
template<>
void JoinHashTable::GetJoinResult<JoinTypes::SEMI>(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    // TODO: 去重
    auto &match = probe_state->match;
    if(match.empty()) {
        return;
    }
    auto &match_result = probe_state->match_result;
    std::vector<std::shared_ptr<arrow::RecordBatch>> results_p;
    for(int i = 0; i < match.size(); i++) {
        std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(match[i], 1);
        results_p.push_back(row);
    }
    result = arrow::ConcatenateRecordBatches(results_p).ValueOrDie();
}

// 外部表为Probe表，返回Probe为false的外部表字段
template<>
void JoinHashTable::GetJoinResult<JoinTypes::ANTI>(std::shared_ptr<arrow::RecordBatch> &chunk, std::shared_ptr<arrow::RecordBatch> &result, std::shared_ptr<ProbeState> &probe_state, std::shared_ptr<ProbeState> &last_probe_state, bool is_final_probe) {
    std::vector<std::shared_ptr<arrow::RecordBatch>> results_p;
    if(last_probe_state != nullptr) {
        auto &bit_map = probe_state->bit_map;
        auto &last_bit_map = last_probe_state->bit_map;
        int size_map = bit_map.size();
        if(!last_bit_map.empty()) {
            if(bit_map.empty()) {
                bit_map = last_bit_map;
            } else {
                for(int i = 0; i < size_map; i++) {
                    bit_map[i] |= last_bit_map[i];
                }
            }
        }
        std::vector<int> no_match;
        if(is_final_probe) {
            for(int i = 0; i < bit_map.size(); i++) {
                if(bit_map[i] == 0) {
                    no_match.push_back(i);
                }
            }
            int no_match_size = no_match.size();
            if(no_match_size != 0) {
                for(int j = 0; j < no_match_size; j++) {
                    std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(no_match[j], 1);
                    results_p.push_back(row);
                }
            } // if(no_match_size != 0)
        } // if(is_final_probe)
    } // if(last_probe_state != nullptr)
    else {
        if(is_final_probe) {
            auto &bit_map = probe_state->bit_map;
            int size_map = bit_map.size();
            std::vector<int> no_match;
            for(int i = 0; i < size_map; i++) {
                if(bit_map[i] == 0) {
                    no_match.push_back(i);
                }
            }
            int no_match_size = no_match.size();
            if(no_match_size != 0) {
                for(int j = 0; j < no_match_size; j++) {
                    std::shared_ptr<arrow::RecordBatch> row = chunk->Slice(no_match[j], 1);
                    results_p.push_back(row);
                }
            } // if(no_match_size != 0)
        } // if(is_final_probe)
    }
    if(results_p.empty()) {
        return;
    }
    result = arrow::ConcatenateRecordBatches(results_p).ValueOrDie();
    // spdlog::info("[{} : {}] ANTI size结果: {}", __FILE__, __LINE__, result->num_rows());
}

std::shared_ptr<arrow::RecordBatch> JoinHashTable::GetSingleRow(Offset offset) {
    int64_t chunk_id = offset.chunk_id;
    int64_t offset_p = offset.offset;
    std::shared_ptr<arrow::RecordBatch> res = data_chunks[chunk_id]->Slice(offset_p, 1);
    return res;
}

void JoinHashTable::GetSingleNullRow() {
    Util::GetSinlgeRowNull(null_row, schema);
}

} // DaseX
